package com.gitee.xhs;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup;

/**
 * @author BigHead
 * @date 2022/6/29 11:05
 * @description EnvironmentConfig
 */
public class EnvironmentConfig {

    private Configuration configuration = new Configuration();

    private Integer parallelism = 1;

    public EnvironmentConfig(Configuration configuration) {
        if (null != configuration) {
            this.configuration = configuration;
        }
    }

    private Boolean useCheckPoint = false;

    private Integer checkpointInterval = 1 * 1000;

    private CheckpointingMode checkpointingMode = CheckpointingMode.EXACTLY_ONCE;

    private String checkPointFilePath = "file:///flink/checkpoint/";

    private ExternalizedCheckpointCleanup externalizedCheckpointCleanup = ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION;

    public void checkpoint(Integer checkpointInterval, CheckpointingMode checkpointingMode, String checkPointFilePath, ExternalizedCheckpointCleanup externalizedCheckpointCleanup) {
        this.useCheckPoint = true;

        if (null != checkpointInterval) {
            this.checkpointInterval = checkpointInterval;
        }

        if (null != checkPointFilePath){
            this.checkPointFilePath = checkPointFilePath;
        }

        if (null != checkpointingMode) {
            this.checkpointingMode = checkpointingMode;
        }

        if (null != externalizedCheckpointCleanup) {
            this.externalizedCheckpointCleanup = externalizedCheckpointCleanup;
        }
    }

    public Configuration getConfiguration() {
        return configuration;
    }

    public Boolean getUseCheckPoint() {
        return useCheckPoint;
    }

    public void setUseCheckPoint(Boolean useCheckPoint) {
        this.useCheckPoint = useCheckPoint;
    }

    public Integer getCheckpointInterval() {
        return checkpointInterval;
    }

    public void setCheckpointInterval(Integer checkpointInterval) {
        this.checkpointInterval = checkpointInterval;
    }

    public Integer getParallelism() {
        return parallelism;
    }

    public void setParallelism(Integer parallelism) {
        this.parallelism = parallelism;
    }

    public CheckpointingMode getCheckpointingMode() {
        return checkpointingMode;
    }

    public void setCheckpointingMode(CheckpointingMode checkpointingMode) {
        this.checkpointingMode = checkpointingMode;
    }

    public String getCheckPointFilePath() {
        return checkPointFilePath;
    }

    public void setCheckPointFilePath(String checkPointFilePath) {
        this.checkPointFilePath = checkPointFilePath;
    }

    public CheckpointConfig.ExternalizedCheckpointCleanup getExternalizedCheckpointCleanup() {
        return externalizedCheckpointCleanup;
    }

    public void setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup externalizedCheckpointCleanup) {
        this.externalizedCheckpointCleanup = externalizedCheckpointCleanup;
    }
}
